package com.example.demo.produce;

import com.example.demo.dto.TestEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;


@Slf4j
@Service
public class KafkaProducerService {

    private final KafkaTemplate<String, Object> kafkaTemplate;



//    @Value("${app.kafka.topics.topic}")
    private String myTopic = "test-topic";

    @Autowired
    public KafkaProducerService(KafkaTemplate<String, Object> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendMessage(TestEvent event) {
        sendMessage(myTopic, event.getKey(), event.getMessage());
    }

    public void sendMessage(String topic, String key,String message) {
        // 发送消息并获取 ListenableFuture
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, key, message);

        // 添加回调处理
        future.addCallback(
                result -> handleSuccess(result),  // 成功回调
                ex -> handleFailure(ex)           // 失败回调
        );
    }

    private void handleSuccess(SendResult<String, Object> result) {
        System.out.println("Sent successfully! " +
                "Topic: " + result.getRecordMetadata().topic() + ", " +
                "Partition: " + result.getRecordMetadata().partition() + ", " +
                "Offset: " + result.getRecordMetadata().offset());
    }

    private void handleFailure(Throwable ex) {
        System.err.println("Failed to send message: " + ex.getMessage());
        // 可在此处重试或记录日志
    }


}